package com.gj.kafka;

import com.gj.feign.TransactionMessageFeign;
import com.gj.model.TransactionMessage;
import com.gj.service.SubstitutionService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

/**
 * author:guijie
 * date:${date}
 * description:TODO
 */
@Component
public class KafkaConsumer {

    @Autowired
    private TransactionMessageFeign feign;
    @Autowired
    private SubstitutionService service;

    @KafkaListener(topics = "transaction")
    public void consumer(ConsumerRecord<String, TransactionMessage> record){
        System.out.println("获得的消息为："+record.value().getMessage());
        try {
            System.out.println("可靠性消息服务消费消息"+record.value().toString());
            //执行修改操作
            boolean flag = service.upd(record.value().getMessage());
            boolean b = feign.confirmCustomerMessage("transaction-substitution", record.value().getId());
            if (b){
                //这里进行手动提交ack
            }
        }catch (Exception e){
            //出现异常则触发重试机制
            throw new RuntimeException(e);
        }
    }

}
